//! 磁盘I/O 相关组件
//!

use crate::{
    pages::RawPageData,
    types::{FullPageID, RelationID}, utils::threads::singleflight::Group,
};
use std::{collections::HashMap, fs::File, os::unix::prelude::{OpenOptionsExt, FileExt}, path::{Path, PathBuf}, convert::identity};
use std::sync::RwLock;

/// 磁盘控制器, 负责IO一个数据库的文件页面IO.
/// 注意, 磁盘控制器不保证同一页面读写的并发安全性,
/// 同一页面读写的并发安全性是由缓存控制器负责的.
pub struct DiskControl {
    /// 放置数据库文件的目录.
    db_dir: PathBuf,

    /// 打开的文件缓存.
    /// 使用读写锁锁定.
    file_cache: RwLock<HashMap<RelationID, File>>,

    /// 用于打开文件消重的控制组.
    single_flight: Group<RelationID, ()>,
}

impl DiskControl {
    pub fn new(db_dir: PathBuf) -> Self {
        DiskControl {
            db_dir,
            file_cache: Default::default(),
            single_flight: Default::default(),
        }
    }

    /// 在linux上, 使用O_DIRECT打开文件.
    #[cfg(target_os = "linux")]
    fn open_page_file<P>(path: P) -> std::io::Result<std::fs::File> 
        where P: AsRef<Path>
    {
        let out = std::fs::File::options()
            .custom_flags(libc::O_DIRECT)
            .create(true)
            .write(true)
            .read(true)
            .open(path)?;
        Ok(out)
    }

    #[cfg(target_os = "macos")]
    fn open_page_file(path: P) -> std::io::Result<std::fs::File> 
        where P: AsRef<Path>
    {
        use std::os::unix::prelude::AsRawFd;
        let out = std::fs::File::options()
            .create(true)
            .write(true)
            .read(true)
            .open(path)?;

        let fd = out.as_raw_fd();
        unsafe {
            let res = libc::fcntl(fd, libc::F_NOCACHE, 1);
            if res != 0 {
                panic!("set page file to F_NOCACHE failed");
            }
        }

        Ok(out)
    }

    fn open_relation_file(mut base_dir: PathBuf, id: RelationID) -> std::io::Result<std::fs::File> {
        let raw: u32 = id.into();
        base_dir.push(format!("{}", raw));
        Self::open_page_file(base_dir)
    }

    pub fn write_page_to(&self, id: FullPageID, buffer: &RawPageData) {
        let relation = id.relation_id();
        let page = id.page_id();
        let offset = page.offset();
        let read_cache = self.file_cache.read().unwrap();
        if let Some(f) = read_cache.get(&relation) {
            f.write_at(buffer, offset as u64).expect("write page failed");
            return;
        }
        drop(read_cache);

        let base_dir = self.db_dir.clone();
        self.single_flight.call(&relation, || {
            let base_dir = identity(base_dir);
            let cache = self.file_cache.read().unwrap();
            if cache.get(&relation).is_some() {
                return;
            };

            // 打开文件
            let f = Self::open_relation_file(base_dir, relation).unwrap();
            self.file_cache.write().unwrap()
                .insert(relation, f);
        });

        let read_cache = self.file_cache.read().unwrap();
        let f = read_cache.get(&relation).unwrap(); 
        f.write_at(buffer, offset as u64).expect("write page failed");

    }

    pub fn read_page_to(&self, id: FullPageID, buffer: &mut RawPageData) {
        let relation = id.relation_id();
        let page = id.page_id();
        let offset = page.offset();
        let read_cache = self.file_cache.read().unwrap();
        if let Some(f) = read_cache.get(&relation) {
            f.read_at(buffer, offset as u64).expect("read page failed");
            return;
        }
        drop(read_cache);

        let base_dir = self.db_dir.clone();
        self.single_flight.call(&relation, || {
            let base_dir = identity(base_dir);
            let cache = self.file_cache.read().unwrap();
            if cache.get(&relation).is_some() {
                return;
            };

            // 打开文件
            let f = Self::open_relation_file(base_dir, relation).unwrap();
            self.file_cache.write().unwrap()
                .insert(relation, f);
        });

        let read_cache = self.file_cache.read().unwrap();
        let f = read_cache.get(&relation).unwrap(); 
        f.read_at(buffer, offset as u64).expect("read page failed");
    }
}
